package org.example.rocketmq.a01_quickstart;

import org.apache.commons.lang3.ThreadUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.example.rocketmq.NamesrvAddr;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * 主动拉取消息
 */
public class ConsumerLitePull {

    public static void main(String[] args) throws Exception {
        litePullConsumer();
    }

    /**
     * 随机获取 queue
     */
    public static void litePullConsumer() throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumer_list_pull_group_01");
//        consumer.setNamesrvAddr("rocketmq.jun.cn:9876");
        consumer.setNamesrvAddr(NamesrvAddr.get());
        consumer.start();
        System.out.printf("Consumer started %n");

        // 指定 topic
        consumer.subscribe("TopicTest", "*");

        // 循环获取消息
        while (true) {
            List<MessageExt> messageExtList = consumer.poll();
            if (messageExtList.isEmpty()) {
                ThreadUtils.sleep(Duration.ofSeconds(1));
                continue;
            }

            messageExtList.forEach(action -> {
                System.out.println(new String(action.getBody()));
                // System.out.println("消息消费成功_" + action);
            });
        }
    }

    /**
     * 指定获取 queue
     */
    public static void litePullConsumerAssign() throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumer_list_pull_group_02");
        consumer.setNamesrvAddr("rocketmq.jun.cn:9876");
        consumer.start();
        System.out.printf("Consumer started %n");

        // 获取消息队列
        Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues("TopicTest");
        consumer.assign(messageQueues);

        // 指定消息队列
        ArrayList<MessageQueue> messageQueuesList = new ArrayList<>(messageQueues);
        consumer.seek(messageQueuesList.get(0), 10);

        // 循环获取消息
        while (true) {
            List<MessageExt> messageExtList = consumer.poll();
            if (messageExtList.isEmpty()) {
                ThreadUtils.sleep(Duration.ofSeconds(1));
                continue;
            }

            messageExtList.forEach(action -> {
                System.out.println(new String(action.getBody()));
                System.out.println("消息消费成功_" + action);
            });
        }
    }

}
